
#include <boost/thread.hpp>

#include <string>


#include "CxTcpClient.h"

using namespace DataRevPart;
using boost::thread;
using std::vector;


//È«¾Öio±äÁ¿,³ÌÐò½áÊøÊ±×¢Òâ¹Ø±Õio
boost::asio::io_service                 CxTcpClient::s_ioservice;
boost::thread_group             		CxTcpClient::s_iothread;
std::vector<CxTcpClient::iowork>        CxTcpClient::s_workgroup;
bool                                    CxTcpClient::s_isioinit=false;
size_t                                  CxTcpClient::s_ioref=0;


//È«¾ÖÍê³É¶Ë¿ÚÖ´ÐÐÏß³Ì
void		CxTcpClient::IoRunFunc()
{
	ERRCODE e;

    s_workgroup.push_back(iowork(new boost::asio::io_service::work(s_ioservice)));

	s_ioservice.run(e);

	if(e)
	{
        CX_ERR_MESSAGE(e.message())
	}  
}

//Ö´ÐÐioº¯Êý
void		CxTcpClient::_RunIo()
{
    size_t processnumber=boost::thread::hardware_concurrency();


    for(int i=0;i<processnumber;++i)
        s_iothread.add_thread(new boost::thread(boost::bind<void>(&CxTcpClient::IoRunFunc)));

    s_isioinit=true;
}

void            CxTcpClient::_ReleaseIO()
{
    size_t processnumber=boost::thread::hardware_concurrency();

    for(int i=0;i<processnumber;++i)
       s_workgroup[i].reset();

    s_ioservice.stop();

    s_isioinit=false;

    s_iothread.join_all();

     cout<<"IO EXIT "<<endl;
}



CxTcpClient::CxTcpClient(const string &portname,UINT portno,_ConnectFunc _func):
			m_portno(portno),m_socket(s_ioservice)
{
    ++s_ioref;
	//Ö´ÐÐÏß³Ì
    if(!s_isioinit)
        _RunIo();
	CX_ASSERT(!portname.empty());

	if(!OpenPort(portname,portno,_func))
		portno=-1;	

	CX_ASSERT(m_socket.is_open());
}



CxTcpClient::CxTcpClient():m_portno(-1),m_socket(s_ioservice)
{
     ++s_ioref;
    if(!s_isioinit)_RunIo();
}


CxTcpClient::~CxTcpClient()
{
	ClosePort();

    if((--s_ioref)==0)
        _ReleaseIO();
}

void	CxTcpClient::ConnectedFunc(const ERRCODE &e)
{
	if(e)
	{
        m_socket.cancel();
        m_socket.close();
		m_portno=-1;
		sig_connect(false);
	}
	else 
	{
		sig_connect(true);
	}
}


/*			Open Port handle*/
bool		CxTcpClient::OpenPort(const string &portname,UINT portno,_ConnectFunc _func)
{
	CX_ASSERT(!portname.empty());

	_Addr		addr=addr.from_string(portname);
	
	if(!addr.is_v4() && !addr.is_v6())
		return false;
	
	_EndPoint	_ep(addr,portno);

    sig_connect.connect(0,boost::bind(_func,_1));

	m_socket.async_connect(_ep,boost::bind<void>(&CxTcpClient::ConnectedFunc,this,
													boost::asio::placeholders::error)
						);

	return true;
}


/*			Close Port Handle	*/
bool		CxTcpClient::ClosePort()
{
	if(!m_socket.is_open())
		return true;

    m_socket.cancel();
	m_socket.close();
	m_portno=-1;

	sig_break();	/*·¢ÉäÖÐ¶ÏÐÅºÅ*/

	CX_ASSERT(!m_socket.is_open());

	return true;
}

inline		bool   CxTcpClient::IsOpen()const
{
	return m_socket.is_open();
}

inline		UINT	CxTcpClient::GetPortState()const
{
	if(IsOpen())
		return PORT_OPENED;
	else return PORT_CLOSED;
}

/*   Direct  Read Port handle*/
size_t		CxTcpClient::ReadPort(BYTE	*_buf,size_t _size)
{
	if(!IsOpen())return 0;
	CX_ASSERT(_buf);
	CX_ASSERT(_size);

	ERRCODE e;
	size_t rtsize=m_socket.read_some(boost::asio::buffer(_buf,_size),e);

	if(e)
	{
		ClosePort();
		HandleErrcode(e);
	}
	return rtsize;
}

/*   Direct  Wirte	Port handle*/
size_t		CxTcpClient::WritePort(	const BYTE *_buf,size_t _size)
{
	if(!IsOpen())return 0;
	CX_ASSERT(_buf);
	CX_ASSERT(_size);

	ERRCODE e;
	size_t rtsize=m_socket.write_some(boost::asio::buffer(_buf,_size),e);

	if(e)
	{
		ClosePort();
		HandleErrcode(e);
	}
	return rtsize;

}


/*Òì²½¶Ë¿Ú¶Á*/
void		CxTcpClient::Async_ReadPort(BYTE* _buf,size_t _size,Async_Handler & handler)
{
	if(!IsOpen())return ;
	CX_ASSERT(_buf);
	CX_ASSERT(_size);

	m_socket.async_read_some(boost::asio::buffer(_buf,_size),
												boost::bind<void>(handler,
															boost::asio::placeholders::error,
															boost::asio::placeholders::bytes_transferred
															)
							);

}

/*Òì²½¶Ë¿ÚÐ´*/
void		CxTcpClient::Async_WritePort(const BYTE *_buf,size_t _size,Async_Handler &handler)
{
	if(!IsOpen())return ;
	CX_ASSERT(_buf);
	CX_ASSERT(_size);

	m_socket.async_write_some(boost::asio::buffer(_buf,_size),
												boost::bind<void>(handler,
															boost::asio::placeholders::error,
															boost::asio::placeholders::bytes_transferred
															)
							);
}





/*´íÎó´¦Àíº¯Êý,¼¯ÖÐ´¦Àí¸÷ÖÖ·µ»Ø´íÎó*/
void	CxTcpClient::HandleErrcode(const ERRCODE &e)
{
	switch(e.value())
	{
	case boost::asio::error::connection_aborted:
		CX_ERR_MESSAGE("Connect Aborted....");
		break;
	case boost::asio::error::connection_refused:
		CX_ERR_MESSAGE("Connect Refuse....");
		break;
	case boost::asio::error::connection_reset:
		CX_ERR_MESSAGE("Connect reset.....");
		break;
	case boost::asio::error::service_not_found:
		CX_ERR_MESSAGE("Service not found");
		break;
	default:
        CX_ERR_MESSAGE(e.message())
		break;
	}
}
